package data_transfer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;
import java.lang.reflect.Field;


public class HDFSToHBase {

    public static String tableName;
    public static String columnFamily;
    public static String inputPath = "hdfs://localhost:9000/hdfstohbase/dat";

    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException {
        tableName = "HDFSToHBase";
        columnFamily = "cf";
        Configuration conf = new Configuration();
        createTable(tableName, columnFamily);
        Job job = Job.getInstance(conf);
        job.setJobName("HDFSToHBase");
        job.setJarByClass(HDFSToHBase.class);
        job.setMapperClass(doMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        TableMapReduceUtil.initTableReducerJob(tableName, doReduce.class, job);
        FileInputFormat.addInputPath(job, new Path(inputPath));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class doMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        private IntWritable one = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().trim().split(" ");
            for (String split : splits) {
                context.write(new Text(split), one);
            }
        }
    }

    public static class doReduce extends TableReducer<Text, IntWritable, NullWritable> {

        private int sum = 0;
        // it's little difficult to pass parameters to map reduce program
        // during execution, for codes are assigned to every host in distributions
        // the only solution is to read parameter stored in hdfs files.

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, NullWritable, Mutation>.Context context) throws IOException, InterruptedException {
            sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            Put put = new Put(Bytes.toBytes(key.toString()));
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("count"),
                    Bytes.toBytes(String.valueOf(sum)));
            context.write(NullWritable.get(), put);
        }
    }

    public static void createTable(String tableName, String... columnFamilies)
            throws IOException {
        Configuration conf = new Configuration();
        Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin();
        TableName tn = TableName.valueOf(tableName);
        TableDescriptor tdDescriptor = null;
        if (admin.isTableAvailable(tn)) {
            System.out.println("table already exists!");
            admin.disableTable(tn);
            admin.deleteTable(tn);
        }
        System.out.println("table does not exist, now creating it...");
        TableDescriptorBuilder tdBuilder = TableDescriptorBuilder.newBuilder(tn);
        for (String cf : columnFamilies) {
            ColumnFamilyDescriptor cfDescriptor = ColumnFamilyDescriptorBuilder.
                    newBuilder(Bytes.toBytes(cf)).build();
            tdBuilder.setColumnFamily(cfDescriptor);
        }
        tdDescriptor = tdBuilder.build();
        admin.createTable(tdDescriptor);
    }
}
